/**
 * 上海中赢金融信息服务有限公司
 * Copyright (c) 2017-2027 Chinazyjr,Inc.All Rights Reserved.
 */

package com.sys.midware.hadoop.hdfs;

import java.io.IOException;
import java.util.ArrayList;
import java.util.HashMap;
import java.util.List;
import java.util.Map;

import org.apache.commons.lang3.ArrayUtils;
import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.BlockLocation;
import org.apache.hadoop.fs.FSDataInputStream;
import org.apache.hadoop.fs.FSDataOutputStream;
import org.apache.hadoop.fs.FileStatus;
import org.apache.hadoop.fs.FileSystem;
import org.apache.hadoop.fs.FileUtil;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.hdfs.DistributedFileSystem;
import org.apache.hadoop.hdfs.protocol.DatanodeInfo;
import org.apache.hadoop.io.IOUtils;

import com.sys.midware.hadoop.exceptions.BHDFSException;
import com.sys.midware.hadoop.hdfs.filter.FilePathFilter;
import com.sys.midware.hadoop.model.BHDFSFile;
import com.sys.midware.hadoop.model.BHDFSFileBlock;
import com.sys.midware.hadoop.model.BHDFSNodeInfo;
import com.sys.midware.logger.Logger;
import com.sys.midware.logger.LoggerFactory;

/**
 * <b>ClassName:</b> Handler.java <br/>
 * <b>Description:</b> TODO <br/>
 * <b>Date:</b> 2016年3月22日 上午11:24:00<br/>
 * 
 * @author mobing
 * @version
 */
public class HDFSHandler {
    private static Logger logger = LoggerFactory.getLogger(HDFSHandler.class);
    
    private static final String CHAR_CODE="UTF-8";

    private Configuration HADOOP_CONFIG = null;

    public HDFSHandler(Configuration HADOOP_CONFIG) {
        this.HADOOP_CONFIG = HADOOP_CONFIG;
    }

    /**
     * 获取FileSystem实例
     * @param  
     * @return  FileSystem
     */
    private FileSystem getFileSystem(){
        FileSystem fs = null;
        try {
            fs = FileSystem.get(this.HADOOP_CONFIG);
        } catch (IOException e) {
            logger.error("fail to getFileSystem!",e);
            throw new BHDFSException("fail to getFileSystem!", e);
        }
        return fs;
    }
    /**
     * 关闭FileSystem
     * @param  
     * @return  void
     */
    private void close(FileSystem fs){
        try {
            fs.close();
        } catch (IOException e) {
            logger.error("FileSystem close error!",e);
            throw new BHDFSException("FileSystem close error!", e);
        }
    }
    
    
    /**
     * 获取data node host name
     * 
     * @param
     * @return List<String>
     */
    public List<BHDFSNodeInfo> getDataNodeHost() {
        List<BHDFSNodeInfo> dataNodeHostList = new ArrayList<BHDFSNodeInfo>();
        DatanodeInfo[] dataNodeStats = null;
        FileSystem fs = null;
        DistributedFileSystem hdfs = null;
        try {
            fs = getFileSystem();
            hdfs = (DistributedFileSystem) fs;
            dataNodeStats = hdfs.getDataNodeStats();
            DatanodeInfo datanodeInfo = null;
            BHDFSNodeInfo nf = null;
            for (int i = 0; i < dataNodeStats.length; i++) {
                datanodeInfo = dataNodeStats[i];
                nf =new BHDFSNodeInfo();
                nf.setHostName(datanodeInfo.getHostName());
                nf.setIPAddr(datanodeInfo.getInfoAddr());
                nf.setCapacity(datanodeInfo.getCapacity());
                nf.setDfsUsed(datanodeInfo.getDfsUsed());
                nf.setRemaining(datanodeInfo.getRemaining());
                dataNodeHostList.add(nf);
            }
        } catch (Exception e) {
            logger.error("get dataNodeStats error!", e);
            throw new BHDFSException("get dataNodeStats error!", e);
        } finally {
            close(fs);
        }
        return dataNodeHostList;
    }

    /**
     * make a new dir in the hdfs the dir may like '/tmp/testdir'
     * @param  
     * @return  void
     */
    public void mkdir(String dir) {
        FileSystem fs = null;
        try {
            fs = getFileSystem();
            fs.mkdirs(new Path(dir));
        } catch (Exception e) {
            logger.error("mkdir error!", e);
            throw new BHDFSException("mkdir error!", e);
        } finally {
            close(fs);
        }

    }

    /**
     * read the hdfs file content notice that the dst is the full path name
     * @param  
     * @return  byte[]
     */
    public byte[] readFile(String filePath) {
        byte[] buffer = null;
        FileSystem fs = null;
        FSDataInputStream is = null;
        try {
            fs =getFileSystem();
            // check if the file exists
            Path path = new Path(filePath);
            if (fs.exists(path)) {
                is = fs.open(path);
                // get the file info to create the buffer
                FileStatus stat = fs.getFileStatus(path);
                // create the buffer
                buffer = new byte[Integer.parseInt(String.valueOf(stat.getLen()))];
                is.readFully(0, buffer);
            }else{
                logger.error("file not found! [filePath="+ filePath +"]"); 
            }
        } catch (Exception e) {
            logger.error("readFile error! [filePath="+ filePath +"]" , e);
            throw new BHDFSException("readFile error![filePath="+ filePath +"]", e);
        } finally {
            close(fs);
            IOUtils.closeStream(is);
        }
        return buffer;
    }

    /**
     * upload the local file to the hds 路径是全路径
     * @param  
     * @return  void
     */
    public void uploadLocalFile2HDFS(String src, String dst) {
        FileSystem fs = null;
        try {
            fs = getFileSystem();
            Path srcPath = new Path(src);
            Path dstPath = new Path(dst);
            fs.copyFromLocalFile(srcPath, dstPath);
        } catch (Exception e) {
            logger.error("uploadLocalFile2HDFS error![", e);
            throw new BHDFSException("uploadLocalFile2HDFS error!", e);
        } finally {
            close(fs);
        }
    }
    /**
     * create a new file in the hdfs. notice that the toCreateFilePath is the
     * full path and write the content to the hdfs file.  
     * @param  
     * @return  void
     */
    public void create(String file, String words) {
        FSDataOutputStream os = null;
        FileSystem fs = null;
        try {
            fs = getFileSystem();
            os = fs.create(new Path(file));
            os.write(words.getBytes(CHAR_CODE));
        } catch (Exception e) {
            logger.error("fail to creat hdfs file!",e);
            throw new BHDFSException("fail to creat hdfs file![filepath=" + file, e);
        } finally {
            close(fs);
            IOUtils.closeStream(os);
        }
    }

    /**
     * delete the hdfs file notice that the dst is the full path name
     * @param  
     * @return  boolean
     */
    public boolean delete(String fPath) {
       return delete(fPath,false);
    }
    /**
     * 是否递归删除文件
     * @param  
     * @return  List<String>
     */
    public boolean delete(String fPath,boolean recursive)  {
        boolean isExists=false;
        FileSystem fs = null;
        try {
            fs = getFileSystem();
            Path path = new Path(fPath);
            isExists=fs.delete(path, recursive);
        } catch (Exception e) {
            logger.error("fail to delete hdfs file![filepath=" + fPath +"]",e);
            throw new BHDFSException("fail to delete hdfs file![filepath=" + fPath+"]", e);
        }finally {
            close(fs);
        }
        return isExists;
    }
    /**
     * 获取文件系统主目录
     */
    public String getHomeDirectory()  {
        FileSystem fs =null;
        Path homeDir= null;
        try {
           fs = getFileSystem();
           homeDir=fs.getHomeDirectory();
        } catch (Exception e) {
            logger.error("fail to getHomeDirectory." ,e);
            throw new BHDFSException("fail to getHomeDirectory.", e);
        }finally{
            close(fs);
        }
        return homeDir==null? null : homeDir.toString();
    }
    /**
     * 获取hdfs文件信息
     * @param
     * @return void
     */
    public BHDFSFile getHDFSFile(String path) {
        FileStatus stats = null;
        BHDFSFile bf = null;
        FileSystem fs =null;
        try {
            fs = getFileSystem();
            stats = fs.getFileStatus(new Path(path));
            //对象转换
            bf = new BHDFSFile();
            bf.setDir(stats.isDirectory());
            bf.setPath(stats.getPath().toString());
            bf.setBlockSize(stats.getBlockSize());
            bf.setOwner(stats.getOwner());
            bf.setLength(stats.getLen());
        } catch (Exception e) {
            logger.error("fail to getHDFSFile.path="+ path,e);
            throw new BHDFSException("fail to getHDFSFile.path="+ path, e);
        } finally {
            close(fs);
        }
        return bf;
    }
    /**
     * 列出hdfs文件
     * @param
     * @return void
     */
    public List<BHDFSFile> listHDFSFile(String path) {
        FileSystem fs =null;
        List<BHDFSFile> bfList = new ArrayList<BHDFSFile>();
        FileStatus[] stats = null;
        try {
            fs = getFileSystem();
            stats = fs.listStatus(new Path(path));
            //对象转换
            BHDFSFile bf = null;
            FileStatus st =null;
            if (stats != null){
                for (int i = 0; i < stats.length; ++i) {
                    st =stats[i];
                    bf = new BHDFSFile();
                    bf.setDir(st.isDirectory());
                    bf.setPath(st.getPath().toString());
                    bf.setBlockSize(st.getBlockSize());
                    bf.setOwner(st.getOwner());
                    bf.setLength(st.getLen());
                    bf.setGroup(st.getGroup());
                    bfList.add(bf);
                }
            }
        } catch (IOException e) {
            logger.error("fail to listHDFSFile!path="+ path,e);
            throw new BHDFSException("fail to listHDFSFile!path="+ path, e);
        } finally {
            close(fs);
        }
        return bfList;
    }

    /**
     * 文件重命名 
     * @param  
     * @return  boolean
     */
    public boolean rename(String oldName, String newName) {
        FileSystem fs =null;
        boolean isok = false;
        try {
            fs = getFileSystem();
            Path oldPath = new Path(oldName);
            Path newPath = new Path(newName);
           isok = fs.rename(oldPath, newPath);
        } catch (Exception e) {
            logger.error("fail to rename![oldName="+ oldName +",newName=" +newName+"]",e);
            throw new BHDFSException("fail to rename![oldName="+ oldName +",newName=" +newName+"]", e);
        }finally { 
            close(fs);
        }
        return isok;
    }
    
    /**
     * 获取文件块所在位置 
     * @param  
     * @return  List<String>
     */
    public List<Map<String,BHDFSFileBlock>> getFileBlockLocations(String filePath)  {
        FileSystem fs =null;
        List<Map<String,BHDFSFileBlock>> mapList = new ArrayList<Map<String,BHDFSFileBlock>>();
        try {
            fs = getFileSystem();
            Path path = new Path(filePath);
            FileStatus filestatus = fs.getFileStatus(path);
            BlockLocation[] blkLocations = fs.getFileBlockLocations(filestatus, 0, filestatus.getLen());
            BHDFSFileBlock fb = null;
            BlockLocation bl =null;
            for(int i=0;i< blkLocations.length;i++){
                Map<String,BHDFSFileBlock> map = new HashMap<String,BHDFSFileBlock>();
                bl = blkLocations[i];
                fb = new BHDFSFileBlock();
               
                fb.setNames(ArrayUtils.toString(bl.getNames()) );
                fb.setOffset(ArrayUtils.toString(bl.getOffset()));
                fb.setLength(ArrayUtils.toString(bl.getLength()));
                fb.setTopologyPaths(ArrayUtils.toString(bl.getTopologyPaths()));
                fb.setHosts(ArrayUtils.toString(bl.getHosts()));
                map.put(String.valueOf(i), fb);
                mapList.add(map);
            }
        }catch (Exception e) {
            logger.error("fail to getFileBlockLocations![filePath="+ filePath +"]",e);
            throw new BHDFSException("fail to getFileBlockLocations![filePath="+ filePath +"]", e);
        }finally{
            close(fs);
        }
        return mapList;
    }
    
    /**
     * 获取文件或目录的修改时间
     * @param  
     * @return  List<String>
     */
    public Long getModificationTime(String fPath) {
        FileSystem fs =null;
        Long modiTime = null;
        try {
            fs = getFileSystem();
            Path path = new Path(fPath);
            FileStatus fileStatus = fs.getFileStatus(path);
             modiTime = fileStatus.getModificationTime();
        } catch (Exception e) {
            logger.error("fail to getModificationTime![fPath="+ fPath +"]",e);
            throw new BHDFSException("fail to getModificationTime![fPath="+ fPath +"]", e);
        }finally{
            close(fs);
        }
        return modiTime;
    }
    /**
     * 校验文件是否存在
     */
    public boolean checkExists(String fPath)  {
        FileSystem fs =null;
        boolean isExists= false;
        try {
            fs = getFileSystem();
            Path path = new Path(fPath);
            isExists = fs.exists(path);
        } catch (Exception e) {
            logger.error("fail to checkExists![fPath="+ fPath +"]",e);
            throw new BHDFSException("fail to checkExists![fPath="+ fPath +"]", e);
        }finally {
            close(fs);
        }
        return isExists;
    }
    
    /** 
     * 对文件路径进行过滤 
     * FileSystema提供globStatus（）方法对文件路径进行过滤，这里的路径必须是hdfs路径 
     *  
     * @param in : 使用通配符 如：hdfs://localhost:9000/hdfs/test/wordcount/in/* 
     * @throws IOException  
     */  
    public String filteFilePaths(String fPath,String matchFile) {  
        FileSystem fs =null;
        StringBuilder sb = new StringBuilder();  
        try {
            fs = getFileSystem();
            //globStatus()方法返回与路径想匹配的所有文件的FileStatus对象数组，并按路径排序。  
            FileStatus[] fss = fs.globStatus(new Path(fPath), new FilePathFilter(matchFile));  
            Path[] paths = FileUtil.stat2Paths(fss);  
            if(paths != null){  
                for(Path path : paths){  
                    sb.append(path.toString() + ",");  
                }  
            }  
            int index = sb.toString().lastIndexOf(",");  
            if(index != -1){  
                return sb.toString().substring(0, index);  
            }  else {
                return null;
            }
        } catch (Exception e) {
            logger.error("fail to filteFilePaths![fPath="+ fPath+",matchFile="+matchFile +"]",e);
            throw new BHDFSException("fail to filteFilePaths![fPath="+ fPath+",matchFile="+matchFile  +"]", e);
        } finally{
            close(fs);
        }
    }  
    
}
